package com.ytktt.study.week12.task4.kafka.producer;

import com.ytktt.study.week12.task4.MessageInfo;
import com.ytktt.study.week12.task4.Order;
import com.ytktt.study.week12.task4.kafka.constant.KafkaConstant;
import com.ytktt.study.week12.task4.redis.constant.RedisConstant;
import com.ytktt.study.week12.task4.util.CurDateStringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.Map;

/**
 * kafka 消息序列生产者拦截器
 *
 * @author ytk
 */
@Slf4j
public class KafkaSeqProducerInterceptor implements ProducerInterceptor<String, Order> {

    private CurDateStringUtil curDateStringUtil;
    private RedisTemplate<String, Object> redisTemplate;

    @Override
    public ProducerRecord<String, Order> onSend(ProducerRecord<String, Order> record) {
        String curDateString = this.curDateStringUtil.getCurDateString();
        String redisKey = String.format(RedisConstant.ORDER_PRODUCER_MESSAGE_SEQ_WITH_PARTITION,
                curDateString, record.partition());
        MessageInfo mi = new MessageInfo();
        mi.setDate(curDateString);
        mi.setSeq(this.redisTemplate.opsForValue().increment(redisKey));
        mi.setPartition(record.partition());
        record.value().setMessageInfo(mi);
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {
        this.curDateStringUtil = (CurDateStringUtil) configs.get(KafkaConstant.CUR_DATE_STRING_UTIL_CONFIG);
        this.redisTemplate = (RedisTemplate<String, Object>) configs.get(KafkaConstant.REDIS_TEMPLATE_CONFIG);
    }
}
